<?php
/**
 * 限流机制
 */
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;

include_once __DIR__.'/config.php';

$exchange_name = 'router';
$queue_name = 'msgs';
$routing_key   = 'key';

/*$exchange_name = 'someExchange';
$routing_key   = 'someRoutking';
$queue_name    = 'someQueue';*/

$args = new AMQPTable();

$args->set('x-max-length', 10);

$connection = new AMQPStreamConnection(HOST,PORT,USER,PASS,VHOST);

$channel = $connection->channel();

$channel->queue_declare($queue_name,false, true, false, false,false,$args);

$channel->queue_bind($queue_name,$exchange_name,$routing_key);

//限流机制开启,参数2000代表当前服务每次只能消费2000条数据，消费完成后继续从MQ中获取2000数据慢慢消费
$channel->basic_qos(null,2000,null);

/**
 * 回调方法
 * @param $message
 */
function process_message($message){
    /*业务逻辑*/

    echo "成功收到消息，消息内容为：{$message->body}".PHP_EOL;

    //消费完消息之后进行应答，告诉rabbitmq我已经消费了消息
    //如果把2000条消息都消费完了，则会在发2000条消息到当前服务器中慢慢消费
    //设置$message->ack()也能消费，不过不能起到一个basic_qos限流的作用
    //$message->ack();
    $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
}

//启动消费者，消费消息
$channel->basic_consume($queue_name,'',false, false, false, false, 'process_message');

//判断是否消费，依次循环打印数据
while ($channel->is_consuming()){
    //消息循环输出到控制台，这里主要输出错误信息之类的
    $channel->wait();
}